package tinynio;

import java.io.IOException;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;

public class SocketCommunicator extends AbstractCommunicator {

	protected Codec codec;
	protected Handler handler;

	private Queue<ByteBuffer> writeBuffer = new LinkedBlockingQueue<ByteBuffer>();

	public SocketCommunicator(Select select) throws IOException {
		this(select, SocketChannel.open());
	}

	public SocketCommunicator(Select select, SelectableChannel sc) throws IOException {
		super(select, sc, SelectionKey.OP_READ);
	}

	@Override
	public void connect(SocketAddress remote) {
		try {
			register();
			channel().connect(remote);

			if (isActive()) {
				handler.onActive(this);
			}
		} catch (IOException e) {
			interest(interestOp | SelectionKey.OP_CONNECT);
		}
	}

	@Override
	public void bind(SocketAddress local) throws IOException {
		channel().socket().bind(local);
	}

	@Override
	public void write(Object obj) throws IOException {
		ByteBuffer bytes = codec.encode(obj);
		if (obj == null)
			return;

		if (!writeBuffer.offer(bytes)) {
			throw new IOException("buffer full!!");
		}

		interest(selectionKey.interestOps() | SelectionKey.OP_WRITE);
		select.getSelector().wakeup();
	}

	@Override
	public void flush() {
		while (true) {
			System.out.println("flush");
			if (writeBuffer.isEmpty()) {
				interest(SelectionKey.OP_READ);
				break;
			}

			try {
				ByteBuffer buffer = writeBuffer.peek();
				int n = 0;
				n = channel().write(buffer);
				System.out.println("n : " + n);
				if (n == 0 && buffer.remaining() > 0) {
					break;
				}

				writeBuffer.remove(buffer);
			} catch (IOException e) {
				handler.onError(this, e);
				break;
			}

		}
	}

	@Override
	public void read() throws IOException {
		ByteBuffer buf = ByteBuffer.allocate(1024);
		int n = 0;

		Object obj;
		while ((n = channel().read(buf)) > 0) {
			buf.flip();
			obj = codec.decode(buf);
			if (obj == null)
				continue;

			handler.onRead(this, obj);
			buf.clear();
		}

		if (n < 0) {
			handler.onClose(this);
			close();
			return;
		}

	}

	@Override
	public SocketChannel channel() {
		return (SocketChannel) super.channel();
	}

	@Override
	public boolean isActive() {
		return channel().socket().isConnected();
	}

	@Override
	public void config(Configuration config) {
		this.handler = config.handler();
		this.codec = config.codec();
		config.configSocket(channel().socket());
	}

	public Handler handler() {
		return handler;
	}

}
